package com.atbeijing.D03;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import scala.Tuple4;

/**
 * 自定义窗口聚合函数,处理滚动时间窗口数据
 * 窗口数据全量聚合:先把窗口所有数据收集起来，等到计算的时候会遍历所有数据
 *
 * 求5s内每个温感器的平均温度,附加窗口信息
 */
public class Example8 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                .keyBy(r -> r.id)//分流
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))//为每条流开一个滚动处理时间窗口,5s
                .process(new WindowResult())//自定义全量聚合函数,适用于每条流的窗口
                .print();
//        每5秒打印一次:Tuple4<分流的key,平均值,窗口开始时间,窗口结束时间>
//        (sensor_0,68.18964937779134,1619524170000,1619524175000)
//        (sensor_9,64.08439761346801,1619524170000,1619524175000)
//        (sensor_8,71.4753715111614,1619524170000,1619524175000)
//        (sensor_7,85.00692678394194,1619524170000,1619524175000)
//        (sensor_6,62.116116188781014,1619524170000,1619524175000)
//        (sensor_5,44.64773448860069,1619524170000,1619524175000)
//        (sensor_4,96.61873238211294,1619524170000,1619524175000)
//        (sensor_3,52.3865345920648,1619524170000,1619524175000)
//        (sensor_2,67.96773296944129,1619524170000,1619524175000)
//        (sensor_1,63.19529092331751,1619524170000,1619524175000)
        env.execute();
    }

    /**
     * 自定义全量聚合
     * ProcessWindowFunction的泛型
     *  <IN>  输入类型 SensorReading
     *  <OUT> 窗口结束后下发的数据类型 Tuple4<分流的key,平均值,窗口开始时间,窗口结束时间>
     *  <KEY> 数据分流的key的类型
     *  <W> Window的子类
     */
    public static class WindowResult extends ProcessWindowFunction<SensorReading, Tuple4<String,Double,Long,Long>, String, TimeWindow>{

        //窗口关闭后执行一次
        //s: 分流的key
        //context: 窗口上下文
        //iterable: 窗口中所有数据集合
        //collector: 下发的集合
        @Override
        public void process(String s, Context context, Iterable<SensorReading> iterable, Collector<Tuple4<String, Double, Long, Long>> collector) throws Exception {
            double sum=0.0;
            int count=0;
            for (SensorReading sensorReading : iterable) {
                sum+=sensorReading.temperature;
                count+=1;
            }
            collector.collect(Tuple4.apply(s,sum/count,context.window().getStart(),context.window().getEnd()));
        }
    }


}
